package org.springboot3.project.demoapi.controller;

import org.apache.dubbo.common.stream.StreamObserver;
import org.apache.dubbo.config.annotation.DubboReference;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springboot3.project.demoapi.StreamDemoService;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.UUID;
import java.util.concurrent.CountDownLatch;

@RestController
public class StreamDemoController {

    private static final Logger log = LoggerFactory.getLogger(StreamDemoController.class);

    @DubboReference(interfaceClass = StreamDemoService.class, version = "1.0", check = false)
    private StreamDemoService streamDemoService;

    @GetMapping("/server-stream")
    public String testServerStream() {
        String requestId = UUID.randomUUID().toString();
        CountDownLatch latch = new CountDownLatch(1);
        StringBuilder result = new StringBuilder();
        streamDemoService.serverStream(requestId, new StreamObserver<String>() {

            @Override
            public void onNext(String data) {
                log.info("receive data: {}", data);
                result.append(data).append("\n");
            }

            @Override
            public void onError(Throwable throwable) {
                log.error("occur error", throwable);
                latch.countDown();
            }

            @Override
            public void onCompleted() {
                log.info("onCompleted");
                latch.countDown();
            }
        });

        try {
            latch.await();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        return result.toString();
    }

    @GetMapping("/client-stream")
    public String testClientStream() {
        CountDownLatch latch = new CountDownLatch(1);

        StreamObserver<String> requestObserver = streamDemoService.clientStream(new StreamObserver<>() {

            @Override
            public void onNext(String data) {
                log.info("get from server : {}", data);
            }

            @Override
            public void onError(Throwable throwable) {
                log.error("occur error", throwable);
                latch.countDown();
            }

            @Override
            public void onCompleted() {
                log.info("onCompleted");
                latch.countDown();
            }
        });
        for (int i = 0; i < 5; i++) {
            requestObserver.onNext("Client message " + i);
        }
        requestObserver.onCompleted();

        try {
            latch.await();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        return "success";
    }
}
